package com.gjy.pulsar.resolve.consumer;

import com.gjy.pulsar.resolve.config.PulsarClientConfig;
import com.gjy.pulsar.resolve.util.JsonUtil;
import lombok.Getter;
import lombok.Setter;
import org.apache.pulsar.client.api.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 手动 ACK 消费者（确保业务成功后才提交消费位点），
 * 并使用 Dead‑Letter Topic 防止因业务异常导致无限重试。
 *
 * @author gjy
 * @version 1.0
 * @since 2025-10-13 15:40:24
 */
public class ManualAckConsumer implements AutoCloseable {
    private static final Logger logger = LoggerFactory.getLogger(ManualAckConsumer.class);

    private final Consumer<byte[]> consumer;

    /**
     * @param topic        订阅的持久化主题
     * @param subscription 订阅名（不同订阅相互独立，支持多消费者并行消费）
     * @throws PulsarClientException 初始化异常
     */
    public ManualAckConsumer(String topic, String subscription) throws PulsarClientException {
        PulsarClient client = PulsarClientConfig.newClient();

        // 死信策略：超过 5 次未成功的消息转入 topic-DLT
        DeadLetterPolicy dlPolicy = DeadLetterPolicy.builder()
                .maxRedeliverCount(5)
                .deadLetterTopic(topic + "-DLT")
                .build();

        this.consumer = client.newConsumer()
                .topic(topic)
                .subscriptionName(subscription)
                // 这里使用 Shared 订阅，多个实例可以并发消费同一分区
                .subscriptionType(SubscriptionType.Shared)
                .subscriptionMode(SubscriptionMode.Durable)
                .deadLetterPolicy(dlPolicy)
                .ackTimeout(30, java.util.concurrent.TimeUnit.SECONDS) // 超时未 ack 自动 redeliver
                .subscribe();

        logger.info("ManualAckConsumer 已创建，topic={}, subscription={}", topic, subscription);
    }

    /**
     * 消费循环（业务自行控制结束）。
     * 业务成功后手动 ack，失败则不 ack（交由自动 redelivery 或 DLT）。
     */
    public void consumeLoop() {
        while (true) {
            try {
                // 拉取单批消息
                Message<byte[]> msg = consumer.receive();
                String key = msg.getKey();
                String body = new String(msg.getData());

                // 业务处理（这里做一个 JSON 反序列化示例）
                try {
                    Order order = JsonUtil.fromJson(body, Order.class);
                    // 业务操作，例如写 DB（必须是幂等的）
                    processOrder(order);
                    // 成功后 ACK，消费进度向前推进
                    consumer.acknowledge(msg);
                    logger.info("订单处理成功，orderId={}, key={}", order.getOrderId(), key);
                } catch (Exception bizEx) {
                    logger.warn("业务处理异常，order={}, 将进入 redelivery", body, bizEx);
                    // 不 ack，交给 Pulsar 自动 redelivery（或达到 maxRedeliverCount 后进入 DLT）
                }

            } catch (PulsarClientException e) {
                logger.error("Consumer receive 异常，可能是网络或关闭导致", e);
                break; // 退出循环，外层可以决定是否重启
            }
        }
    }

    private void processOrder(Order order) {
        // TODO: 真实业务代码（必须实现幂等，如唯一约束）
    }

    @Override
    public void close() {
        try {
            consumer.close();
            logger.info("ManualAckConsumer 已关闭");
        } catch (Exception e) {
            logger.warn("关闭 consumer 时出现异常", e);
        }
    }

    /**
     * 示例业务对象（可自行替换为自己的 POJO）
     */
    @Setter
    @Getter
    public static class Order {
        // getters / setters
        private String orderId;
        private double amount;

    }
}